package cn.tecnova.cd

import cn.tecnova.utils.ConfigHandler
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Dataset, SQLContext}

/**
  * description:
  * jdbc:mysql://www.slave4.com:3306/tec_base2
  * qichacha_search_menlo
  * 6884 7314
  **/
object BaseQichachaSearchMenlo {
  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      //      .setMaster("local[*]")
      .set("es.index.auto.create", "true")
      .set("es.nodes", ConfigHandler.esNodes)
      .set("es.port", ConfigHandler.esPort)
      .set("es.nodes.wan.only", "true")

    val sc = new SparkContext(conf)
    val sqLContext = new SQLContext(sc)

    import sqLContext.implicits._
    val rowDF = sqLContext.read.jdbc("jdbc:mysql://www.slave4.com:3306/tec_base2", "qichacha_search_menlo", ConfigHandler.props2)


    //导入写es相关的包
    import org.elasticsearch.spark.sql._

    val result: Dataset[BaseQichachaSearchMenlo] = rowDF.map(row => {
      BaseQichachaSearchMenlo(
        row.getAs[Long]("id"),
        row.getAs[String]("KeyNo"),
        row.getAs[String]("Name"),
        row.getAs[String]("No"),
        row.getAs[String]("BelongOrg"),
        row.getAs[String]("OperName"),
        row.getAs[String]("StartDate"),
        row.getAs[String]("EndDate"),
        row.getAs[String]("Status"),
        row.getAs[String]("Province"),
        row.getAs[String]("UpdatedDate"),
        row.getAs[String]("ShortStatus"),
        row.getAs[String]("CreditCode"),
        row.getAs[String]("RegistCapi"),
        row.getAs[String]("EconKind"),
        row.getAs[String]("Address"),
        row.getAs[String]("IndustryCode"),
        row.getAs[String]("SubIndustryCode"),
        row.getAs[String]("Industry"),
        row.getAs[String]("SubIndustry"),
        row.getAs[String]("Scope"),
        row.getAs[String]("ContactNumber"),
        row.getAs[String]("Email"),
        row.getAs[String]("WebSite"),
        row.getAs[String]("ImageUrl"),
        row.getAs[String]("OrgNo"),
        row.getAs[String]("EnglishName"),
        row.getAs[String]("Product"),
        row.getAs[Int]("Type"),
        row.getAs[String]("OriginalName"),
        row.getAs[String]("HitField"),
        row.getAs[String]("HitValue"),
        row.getAs[String]("mongo_id")
      )
    })
    result.saveToEs("qichacha_search_menlo" + "/qichacha_search_menlo")


    sc.stop()

  }

}

case class BaseQichachaSearchMenlo(

                                var id:Long,
                                var keyNo:String,
                                var name:String,
                                var no:String,
                                var belongOrg:String,
                                var operName:String,
                                var startDate:String,
                                var endDate:String,
                                var status:String,
                                var province:String,
                                var updatedDate:String,
                                var shortStatus:String,
                                var creditCode:String,
                                var registCapi:String,
                                var econKind:String,
                                var address:String,
                                var industryCode:String,
                                var subIndustryCode:String,
                                var industry:String,
                                var subIndustry:String,
                                var scope:String,
                                var contactNumber:String,
                                var email:String,
                                var webSite:String,
                                var imageUrl:String,
                                var orgNo:String,
                                var englishName:String,
                                var product:String,
                                var Type:Int,
                                var originalName:String,
                                var hitField:String,
                                var hitValue:String,
                                var mongoId:String

                              )
